package cn.csg.ams.collect2.poptopics.controller;

import cn.csg.ams.collect2.poptopics.datahub.TupleRecordDataUtils;
import cn.csg.ams.collect2.poptopics.service.MkPrPfOutageEventDetailService;
import com.aliyun.datahub.client.model.RecordEntry;
import com.aliyun.datahub.client.model.TupleRecordData;
import com.aliyun.datahub.clientlibrary.config.ProducerConfig;
import com.aliyun.datahub.clientlibrary.producer.Producer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.UUID;

/**
 * @author: zhanggang
 * @ClassName: MkPrPfOutageEventDetailController
 * @date: 2023-11-17 09:06:55
 * @Description: 停电事件明细（flink计算出来的数据基表）(mk_pr_pf_outage_event_detail)表控制层
 */
@RestController
@RequestMapping("/member/V3/mkPrPfOutageEventDetail")
public class MkPrPfOutageEventDetailController {
    /**
     * 服务对象
     */
    @Autowired
    private MkPrPfOutageEventDetailService mkPrPfOutageEventDetailService;

    @Value("${datahub.endpoint}")
    private String endpoint;

    @Value("${datahub.accessId}")
    private String accessId;

    @Value("${datahub.accessKey}")
    private String accessKey;


    private static String projectName = "jlxt_syy_pro_datahub";


    private static String topicName = "topic_ods_0300domk_mc_term_alarm_r";

    @Autowired
    private JdbcTemplate jdbcTemplate;


    /**
     * 推送数据到datahub
     */
    @PostMapping("/pushDatahub")
    public void pushDatahub(@RequestBody Map<String, Object> param) {
        String eventId = UUID.randomUUID().toString().replace("-", "");
        param.put("alarm_id", eventId);

        String preReceTime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(System.currentTimeMillis() - 30000L));
        String alarmTime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(System.currentTimeMillis() - 35000L));
        param.put("pre_rece_time", preReceTime);
        param.put("alarm_time", alarmTime);

        ProducerConfig config = new ProducerConfig(endpoint, accessId, accessKey);
        Producer producer = new Producer(projectName, topicName, config);

        List<RecordEntry> records = getRecords(param);
        producer.send(records, "0", 100);
    }


    private List<RecordEntry> getRecords(Map<String, Object> param) {
        List<RecordEntry> records = new ArrayList<>();
        RecordEntry entry = new RecordEntry();
        TupleRecordData recordData = new TupleRecordData(TupleRecordDataUtils.getRecordSchema());
        param.forEach(recordData::setField);
        entry.setRecordData(recordData);
        records.add(entry);
        return records;
    }

//{
//     "run_term_id": "1111111120806004",
//     "run_meter_id": "1511111145528878",
//     "measure_point_no": 0,
//     "alarm_code": "E200003301",
//     "alarm_sour_code": "2",
//     "pre_rece_time": "20231220120000",
//     "alarm_time": "20231220120000"
// }


    public void test(Producer producer, int offset) {
        // List<Map<String, Object>> list = jdbcTemplate.queryForList(
        //         "select * \n" +
        //                 "from mk_mc_term_alarm_log_test \n" +
        //                 "order by alarm_time, run_meter_id, alarm_code\n" +
        //                 "limit 1000 offset " + offset
        // );

        // List<Map<String, Object>> list = jdbcTemplate.queryForList(
        //         "select alarm_id, run_term_id, run_meter_id, measure_point_no, alarm_code, alarm_ext_data, is_fals, is_rest, alarm_name, alarm_level_code, alarm_type_code, alarm_sour_code, alarm_class, alarm_creat_way, is_disp, proc_pers, terminal_type, work_order_mode, order_id, order_state, statute_alarm_code, area_code, pre_rece_time, alarm_time, rest_time from mk_mc_term_alarm_log order by alarm_time, alarm_id limit 1000 offset " + offset
        // );

        List<Map<String, Object>> list = jdbcTemplate.queryForList(
                "select\n" +
                        "\talarm_id, run_term_id, run_meter_id, measure_point_no::bigint as measure_point_no, alarm_code, alarm_ext_data, is_fals, is_rest, \n" +
                        "\talarm_name, alarm_level_code, alarm_type_code, alarm_sour_code, alarm_class, alarm_creat_way, \n" +
                        "\tis_disp, proc_pers, terminal_type, work_order_mode, order_id, \n" +
                        "\torder_state, statute_alarm_code, \n" +
                        "\tconcat(area_code, '66') as area_code, \n" +
                        "\tto_char(pre_rece_time, 'YYYYMMDDHH24MISS') as pre_rece_time, \n" +
                        "\tto_char(alarm_time, 'YYYYMMDDHH24MISS') as alarm_time, \n" +
                        "\tto_char(rest_time, 'YYYYMMDDHH24MISS') as rest_time\n" +
                        "from mk_mc_term_alarm \n" +
                        "where data_part = 20240315\n" +
                        "and alarm_code in ('E200003301', 'E200003401')\n" +
                        "and area_code = '031900'\n" +
                        "and load_time is not null\n" +
                        "order by alarm_time\n" +
                        "limit 1000 offset " + offset
        );


        // List<Map<String, Object>> list = jdbcTemplate.queryForList(
        //         "select * from mk_mc_term_alarm_log_test where run_term_id in ('1111111129884868') \n" +
        //                 "and alarm_code = 'E200003301'\n" +
        //                 "order by alarm_time"
        // );

        List<RecordEntry> records = new ArrayList<>();
        for (Map<String, Object> map : list) {
            RecordEntry entry = new RecordEntry();
            TupleRecordData recordData = new TupleRecordData(TupleRecordDataUtils.getRecordSchema());
            map.forEach(recordData::setField);
            recordData.setField(0, UUID.randomUUID().toString().replace("-", ""));
            entry.setRecordData(recordData);
            records.add(entry);
        }

        System.out.println(records.size());
        producer.send(records, "1", 100);
    }


    // @PostConstruct
    public void test66() {
        ProducerConfig config = new ProducerConfig(endpoint, accessId, accessKey);
        Producer producer = new Producer(projectName, topicName, config);

        // test(producer, 0);

        int offset = 0;
        while (offset <= 97000) {
            test(producer, offset);
            offset += 1000;
        }
    }
}

